package cn.pengpeng.dmp.report

import cn.pengpeng.dmp.beans.OffLineReport.ReportAreaAnalysis
import cn.pengpeng.dmp.utils.{ConfigHandler, RptKpiTools}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

/**
  * Created by root on 2018/12/5.
  */
object RptAreaAnalysisCore {
  def main(args: Array[String]): Unit = {
    val inputPath = "d:\\data\\spark\\out1"
    val conf = new SparkConf()
      .setAppName("RptAreaAnalysisCore")
      .setMaster("local[*]")
    val spark: SparkSession = SparkSession
      .builder()
      .config(conf)
      .getOrCreate()
    val parquet: DataFrame = spark.read.parquet(inputPath)

    import spark.implicits._

    val result: DataFrame = parquet.rdd.map(row => {
      val pname = row.getAs[String]("provincename")
      val cname = row.getAs[String]("cityname")
      ((pname, cname), RptKpiTools.offLineKpi(row))
    }).reduceByKey((list1, list2) => {
      val zip: List[(Double, Double)] = list1.zip(list2)
      val map: List[Double] = zip.map(z => z._1 + z._2)
      map
    }).map(rs => ReportAreaAnalysis(rs._1._1, rs._1._2, rs._2(0), rs._2(1), rs._2(2), rs._2(3), rs._2(4), rs._2(5), rs._2(6), rs._2(7), rs._2(8)))
      .toDF
    result.write.mode(SaveMode.Overwrite).jdbc(ConfigHandler.url,ConfigHandler.areaAnalysis_table,ConfigHandler.dbProps)

    spark.stop()
  }

}
